/*
 *       Copyright (c) 2009 All rights reserved
 *               Asemantics S.r.l
 *
 * Redistribution and use in source and binary forms, with or without
 * modification, are permitted provided that the following conditions
 * are met:
 *
 * 1. Redistributions of source code must retain the above copyright
 *    notice, this list of conditions and the following disclaimer.
 *
 * 2. Redistributions in binary form must reproduce the above copyright
 *    notice, this list of conditions and the following disclaimer in
 *    the documentation and/or other materials provided with the
 *    distribution.
 *
 * 3. The end-user documentation included with the redistribution,
 *    if any, must include the following acknowledgment:
 *       "This product includes software developed by
 *        Asemantics S.r.l."
 *    Alternately, this acknowledgment may appear in the software itself,
 *    if and wherever such third-party acknowledgments normally appear.
 *
 * 4. All advertising materials mentioning features or use of this software
 *    must display the following acknowledgement:
 *    This product includes software developed by Asemantics S.r.l.
 *    the Semantic Web company, Rome, London, Leiden and its contributors.
 *
 * 5. Neither the name of the company nor the names of its contributors
 *    may be used to endorse or promote products derived from this software
 *    without specific prior written permission.
 *
 * THIS SOFTWARE IS PROVIDED BY THE REGENTS AND CONTRIBUTORS ``AS IS'' AND
 * ANY EXPRESSED OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
 * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR
 * PURPOSE ARE DISCLAIMED.  IN NO EVENT SHALL THE REGENTS OR CONTRIBUTORS BE LIABLE
 * FOR ANY DIRECT, INDIRECT, INCIDENTAL,
 * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT
 * NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES;
 * LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION)
 * HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT,
 * STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
 * ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED
 * OF THE POSSIBILITY OF SUCH DAMAGE.
 *
 */
package com.asemantics.memos2;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.util.List;
import java.util.Set;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.logging.Level;
import java.util.logging.Logger;

import net.spy.memcached.AddrUtil;
import net.spy.memcached.ConnectionFactory;
import net.spy.memcached.DefaultConnectionFactory;
import net.spy.memcached.MemcachedClient;

import org.apache.commons.codec.digest.DigestUtils;

import com.opensymphony.oscache.base.Config;
import com.opensymphony.oscache.base.persistence.CachePersistenceException;
import com.opensymphony.oscache.base.persistence.PersistenceListener;

/**
 * A memcached based {@link PersistenceListener} for OSCache.
 *
 * To enable it, just set the <code>cache.persistence.class</code> property in
 * the classpath resource <code>oscache.properties</code> as:
 *
 * <pre>cache.persistence.class=com.asemantics.memos2.MemcachedListener</pre>
 *
 * Follow below supported properties users can add to <code>oscache.properties</code>
 * classpath resource to configure the listener:
 * <table>
 * <tr>
 * <th>property</th>
 * <th>description</th>
 * <th>default</th>
 * </tr>
 * <tr>
 * <td><code>cache.memcached.servers</code></td>
 * <td>the <code>${server}:${port}</code> connection string list</td>
 * <td><code>localhost:11211</code></td>
 * </tr>
 * <tr>
 * <td><code>cache.memcached.connectionfactory</code></td>
 * <td>The <code>net.spy.memcached.ConnectionFactory</code> class</td>
 * <td><code>net.spy.memcached.DefaultConnectionFactory</code></td>
 * </tr>
 * <tr>
 * <td><code>cache.memcached.keyprefix</code></td>
 * <td>The key prefix, useful when memcached server are shared between more
 * than one client, to avoid keys collision</td>
 * <td><code>_oscache_</code></td>
 * </tr>
 * <tr>
 * <td><code>cache.memcached.groupprefix</code></td>
 * <td>Same for key, but to identify groups; a group is generally stored using
 * <code>cache.memcached.keyprefix</code>+<code>cache.memcached.groupprefix</code>
 * concatenation</td>
 * <td><code>__group__</code></td>
 * </tr>
 * <tr>
 * <td><code>cache.memcached.expiration</code></td>
 * <td>The number of seconds entries will be stored</td>
 * <td>the number of seconds in 30 days</td>
 * </tr>
 * <tr>
 * <td><code>cache.memcached.asyncget</code></td>
 * <td>Flag to denote if client uses sync/async get</td>
 * <td>true</td>
 * </tr>
 * <tr>
 * <td><code>cache.memcached.timeout</code></td>
 * <td>The connection timeout for async operations (if set)</td>
 * <td><code>5</code></td>
 * </tr>
 * <tr>
 * <td><code>cache.memcached.timeoutunit</code></td>
 * <td>The connection timeout unit for async operations</td>
 * <td><code>SECONDS</code>, see {@link TimeUnit} to see admitted values</td>
 * </tr>
 * </table>
 *
 * @author Simone Tripodi (simone.tripodi)
 * @version $Id: MemcachedListener.java 18 2009-05-24 00:31:44Z simone.tripodi $
 */
@SuppressWarnings("unchecked")
public final class MemcachedListener implements PersistenceListener {

    /**
     * The <i>cache.memcached.servers</i> string constant.
     */
    private static final String SERVERS = "cache.memcached.servers";

    /**
     * The default memcached connection string, <code>localhost:11211</code>.
     */
    private static final String DEFAULT_SERVERS = "localhost:11211";

    /**
     * The <i>cache.memcached.connectionfactory</i> string constant.
     */
    private static final String CONNECTION_FACTORY = "cache.memcached.connectionfactory";

    /**
     * The default {@link ConnectionFactory} class.
     */
    private static final Class<? extends ConnectionFactory> DEFAULT_CONNECTION_FACTORY = DefaultConnectionFactory.class;

    /**
     * The <i>cache.memcached.keyprefix</i> string constant.
     */
    private static final String KEY_PREFIX = "cache.memcached.keyprefix";

    /**
     * The default key prefix, <code>_oscache_</code>.
     */
    private static final String DEFAULT_KEY_PREFIX = "_oscache_";

    /**
     * The <i>cache.memcached.groupprefix</i> string constant.
     */
    private static final String GROUP_PREFIX = "cache.memcached.groupprefix";

    /**
     * The default key prefix, <code>__group__</code>.
     */
    private static final String DEFAULT_GROUP_PREFIX = "__group__";

    /**
     * The <i>cache.memcached.expiration</i> string constant.
     */
    private static final String EXPIRATION = "cache.memcached.expiration";

    /**
     * The default memcached entry expiration, the number of seconds in 30 days.
     */
    private static final int DEFAULT_EXPIRATION = 60 * 60 * 24 * 30;

    /**
     * The <i>cache.memcached.timeout</i> string constant.
     */
    private static final String TIMEOUT = "cache.memcached.timeout";

    /**
     * The default timeout value, <code>5</code>.
     */
    private static final int DEFAULT_TIMEOUT = 5;

    /**
     * The <i>cache.memcached.timeoutunit</i> string constant.
     */
    private static final String TIMEOUT_UNIT = "cache.memcached.timeoutunit";

    /**
     * The default timeout unit, <code>SECONDS</code>.
     */
    private static final TimeUnit DEFAULT_TIMEOUT_UNIT = TimeUnit.SECONDS;

    /**
     * The <i>cache.memcached.asyncget</i> string constant.
     */
    private static final String ASYNC_GET = "cache.memcached.asyncget";

    /**
     * The default flag to use the asynchronous memcached get, <code>true</code>.
     */
    private static final boolean DEFAULT_ASYNC_GET = true;

    /**
     * This class logger.
     */
    private final Logger logger = Logger.getLogger(this.getClass().getName());

    /**
     * The memcached client reference.
     */
    private MemcachedClient memcachedClient;

    /**
     * The key prefix reference.
     */
    private String keyPrefix;

    /**
     * The group prefix reference.
     */
    private String groupPrefix;

    /**
     * The flag to switch from sync to async memcached get.
     */
    private boolean usingAsyncGet;

    /**
     * The memcached entries expiration time.
     */
    private int expiration;

    /**
     * The memcached connection timeout.
     */
    private int timeout;

    /**
     * The timeout unit.
     */
    private TimeUnit timeUnit;

    /**
     * {@inheritDoc}
     */
    public void clear() throws CachePersistenceException {
        // TODO is there a way to clean and disconnect the memcached client?!?
    }

    /**
     * {@inheritDoc}
     */
    public PersistenceListener configure(Config config) {
        ConnectionFactory connectionFactory = this.getConnectionFactory(config);

        String servers = this.getStringProperty(config, SERVERS, DEFAULT_SERVERS);
        List<InetSocketAddress> addresses = AddrUtil.getAddresses(servers);

        this.keyPrefix = this.getStringProperty(config, KEY_PREFIX, DEFAULT_KEY_PREFIX);
        this.groupPrefix = this.getStringProperty(config, GROUP_PREFIX, DEFAULT_GROUP_PREFIX);
        this.expiration = this.getIntProperty(config, EXPIRATION, DEFAULT_EXPIRATION);
        this.timeout = this.getIntProperty(config, TIMEOUT, DEFAULT_TIMEOUT);
        this.timeUnit = this.getTimeUnitProperty(config);
        this.usingAsyncGet = this.getBooleanProperty(config, ASYNC_GET, DEFAULT_ASYNC_GET);

        if (this.logger.isLoggable(Level.FINE)) {
            this.logger.fine("Creating new memcached client using configuration: { servers='"
                    + servers
                    +"', connectionFactory='"
                    + connectionFactory.getClass().getName()
                    + "', keyPrefix='"
                    + this.keyPrefix
                    + "', groupPrefix='"
                    + this.groupPrefix
                    + "', timeout='"
                    + this.timeout
                    + "', timeUnit='"
                    + this.timeUnit
                    + "', usingAsyncGet='"
                    + this.usingAsyncGet
                    + "' }");
        }

        try {
            this.memcachedClient = new MemcachedClient(connectionFactory, addresses);
        } catch (IOException e) {
            String message = "Impossible to instantiate a new memecached client instance, see nested exceptions";
            if (this.logger.isLoggable(Level.SEVERE)) {
                this.logger.log(Level.SEVERE, message, e);
            }
            throw new RuntimeException(message, e);
        }

        return this;
    }

    /**
     * {@inheritDoc}
     */
    public boolean isGroupStored(String groupName) throws CachePersistenceException {
        return this.retrieveGroup(groupName) != null;
    }

    /**
     * {@inheritDoc}
     */
    public boolean isStored(String key) throws CachePersistenceException {
        return this.retrieve(key) != null;
    }

    /**
     * {@inheritDoc}
     */
    public void remove(String key) throws CachePersistenceException {
        String hashedKey = this.buildHashedKey(key);

        if (this.logger.isLoggable(Level.FINE)) {
            this.logger.fine("Removing cached object with key='"
                    + key
                    + "' hashedKey='"
                    + hashedKey
                    + "'");
        }

        this.memcachedClient.delete(key);
    }

    /**
     * {@inheritDoc}
     */
    public void removeGroup(String groupName) throws CachePersistenceException {
        String hashedGroupName = this.buildHashedGroupKey(groupName);

        if (this.logger.isLoggable(Level.FINE)) {
            this.logger.fine("Looking for cached gruop with groupName='"
                    + groupName
                    + "' hashedGroupName='"
                    + hashedGroupName
                    + "'");
        }

        this.memcachedClient.delete(hashedGroupName);
    }

    /**
     * {@inheritDoc}
     */
    public Object retrieve(String key) throws CachePersistenceException {
        String hashedKey = this.buildHashedKey(key);

        if (this.logger.isLoggable(Level.FINE)) {
            this.logger.fine("Looking for cached object with key='"
                    + key
                    + "' hashedKey='"
                    + hashedKey
                    + "' using async get="
                    + this.usingAsyncGet);
        }

        return this.retrieveFromMemcached(hashedKey);
    }

    /**
     * {@inheritDoc}
     */
    public Set retrieveGroup(String groupName) throws CachePersistenceException {
        String hashedGroupName = this.buildHashedGroupKey(groupName);

        if (this.logger.isLoggable(Level.FINE)) {
            this.logger.fine("Looking for cached group with groupName='"
                    + groupName
                    + "' hashedGroupName='"
                    + hashedGroupName
                    + "'");
        }

        return (Set) this.retrieveFromMemcached(hashedGroupName);
    }

    /**
     * {@inheritDoc}
     */
    public void store(String key, Object obj) throws CachePersistenceException {
        String hashedKey = this.buildHashedKey(key);
        this.storeInMemcached(hashedKey, obj);
    }

    /**
     * {@inheritDoc}
     */
    public void storeGroup(String groupName, Set group) throws CachePersistenceException {
        String hashedGroupName = this.buildHashedGroupKey(groupName);
        this.storeInMemcached(hashedGroupName, group);
    }

    /**
     * 
     * @param config
     * @param propName
     * @param defaultValue
     * @return
     */
    private String getStringProperty(final Config config, final String propName,
            final String defaultValue){
        String value = config.getProperty(propName);
        if (value != null) {
            if (this.logger.isLoggable(Level.FINE)) {
                this.logger.fine("Retrieved property '"
                        + propName
                        + "' from the config with value '"
                        + value
                        + "'");
            }

            return value;
        }

        if (this.logger.isLoggable(Level.FINE)) {
            this.logger.fine("Property '"
                    + propName
                    + "' not specified in the config, using default value '"
                    + defaultValue
                    + "'");
        }

        return defaultValue;
    }

    /**
     * 
     * @param config
     * @param propName
     * @param defaultValue
     * @return
     */
    private boolean getBooleanProperty(final Config config, final String propName,
            final boolean defaultValue) {
        String value = getStringProperty(config, propName, null);
        try {
            if (value != null) {
                return Boolean.parseBoolean(value);
            }
        } catch (Exception e){
            if (this.logger.isLoggable(Level.SEVERE)) {
                this.logger.log(Level.SEVERE, "Error parsing value '"
                        + value
                        + "', using default value '"
                        + defaultValue
                        + "'", e);
            }
        }
        return defaultValue;
    }

    /**
     * 
     * @param config
     * @param propName
     * @param defaultValue
     * @return
     */
    private int getIntProperty(final Config config, final String propName,
            final int defaultValue) {
        String value = getStringProperty(config, propName, null);
        try {
            if (value != null) {
                return Integer.parseInt(value);
            }
        } catch (Exception e){
            if (this.logger.isLoggable(Level.SEVERE)) {
                this.logger.log(Level.SEVERE, "Error parsing value '"
                        + value
                        + "', using default value '"
                        + defaultValue
                        + "'", e);
            }
        }
        return defaultValue;
    }

    /**
     * 
     * @param config
     * @return
     */
    private TimeUnit getTimeUnitProperty(final Config config) {
        String value = this.getStringProperty(config, TIMEOUT_UNIT, null);

        try {
            if (value != null) {
                return TimeUnit.valueOf(value.toUpperCase());
            }
        } catch (Exception e) {
            if (this.logger.isLoggable(Level.SEVERE)) {
                this.logger.log(Level.SEVERE, "TimeUnit '"
                        + value
                        + "' doesn't exist, using default value '"
                        + DEFAULT_TIMEOUT_UNIT
                        + "'", e);
            }
        }

        return DEFAULT_TIMEOUT_UNIT;
    }

    /**
     * 
     * @param config
     * @return
     */
    private ConnectionFactory getConnectionFactory(final Config config) {
        Class<?> clazz;
        String className = this.getStringProperty(config, CONNECTION_FACTORY, null);

        if (className == null) {
            if (this.logger.isLoggable(Level.FINE)) {
                this.logger.fine("Property '"
                        + CONNECTION_FACTORY
                        + "' not found in the config, using default class");
            }
            clazz = DEFAULT_CONNECTION_FACTORY;
        }

        try {
            clazz = Class.forName(className);

            if (!ConnectionFactory.class.isAssignableFrom(clazz)) {
                if (this.logger.isLoggable(Level.SEVERE)) {
                    this.logger.log(Level.SEVERE, "Specified class '"
                            + className
                            + "' is not a '"
                            + ConnectionFactory.class.getName()
                            + "' implementation, using default class");
                }
                clazz = DEFAULT_CONNECTION_FACTORY;
            }
        } catch (ClassNotFoundException e) {
            if (this.logger.isLoggable(Level.SEVERE)) {
                this.logger.log(Level.SEVERE, "Impossible to load specified class '"
                        + className
                        + "', using default class", e);
            }
            clazz = DEFAULT_CONNECTION_FACTORY;
        }

        try {
            return (ConnectionFactory) clazz.newInstance();
        } catch (Exception e) {
            String message = "Impossible to instantiate '"
                + clazz.getName()
                + "', see nested exceptions";

            if (this.logger.isLoggable(Level.SEVERE)) {
                this.logger.log(Level.SEVERE, message, e);
            }

            throw new RuntimeException(message, e);
        }
    }

    /**
     * 
     * @param key
     * @return
     */
    private String buildHashedKey(final String key) {
        return this.keyPrefix + DigestUtils.md5Hex(key);
    }

    /**
     * 
     * @param group
     * @return
     */
    private String buildHashedGroupKey(final String groupName) {
        return this.keyPrefix + this.groupPrefix + DigestUtils.md5Hex(groupName);
    }

    /**
     * 
     * @param hashedKey
     * @return
     * @throws CachePersistenceException
     */
    private final Object retrieveFromMemcached(final String hashedKey) throws CachePersistenceException {
        Object obj = null;

        if (this.usingAsyncGet) {
            Future<Object> future = this.memcachedClient.asyncGet(hashedKey);
            try {
                obj = future.get(this.timeout, this.timeUnit);
            } catch (Exception e) {
                future.cancel(false);
                String errorMessage = "Impossible to retrieve object from memcache identified by '"
                    + hashedKey
                    + "'";

                if (this.logger.isLoggable(Level.SEVERE)) {
                    this.logger.log(Level.SEVERE, errorMessage, e);
                }

                throw new CachePersistenceException(errorMessage);
            }
        } else {
            obj = this.memcachedClient.get(hashedKey);
        }

        return obj;
    }

    /**
     * 
     * @param hashedKey
     * @param obj
     */
    private void storeInMemcached(final String hashedKey, final Object obj) {
        if (this.logger.isLoggable(Level.FINE)) {
            this.logger.fine("Storing object ('"
                    + hashedKey
                    + "', '"
                    + obj
                    + "') with timeout="
                    + this.expiration);
        }

        Future<Boolean> future = this.memcachedClient.set(hashedKey, this.expiration, obj);
        if (this.logger.isLoggable(Level.FINE)) {
            try {
                boolean status = future.get(this.timeout, this.timeUnit);

                this.logger.fine("Status after set object identified by '"
                            + hashedKey
                            + "' is '"
                            + status
                            + "'");
            } catch (Exception e) {
                future.cancel(false);

                if (this.logger.isLoggable(Level.SEVERE)) {
                    this.logger.log(Level.SEVERE,
                            "Impossible to know status after setting object identified by '"
                            + hashedKey
                            + "'", e);
                }
            }
        }
    }

}
